package spc.webos.queue;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.log4j.Logger;

public class BizPoolOnMessage implements IOnMessage {
	protected Logger log = Logger.getLogger(super.getClass());
	protected ExecutorService bizpool;
	protected IOnMessage onMessage;
	protected int maxBlockSize = -1;

	public BizPoolOnMessage(int size, IOnMessage onMessage) {
		this.bizpool = Executors.newFixedThreadPool(size);
		this.onMessage = onMessage;
	}

	public BizPoolOnMessage(int size, IOnMessage onMessage, int maxBlockSize) {
		this.bizpool = Executors.newFixedThreadPool(size);
		this.onMessage = onMessage;
		this.maxBlockSize = maxBlockSize;
	}

	public BizPoolOnMessage() {
	}

	public void onMessage(Object obj, AccessTPool pool, AbstractReceiverThread thread) throws Exception {
		this.log.debug("add OnMessage thread...");
		if (this.maxBlockSize >= 0)
			isBlock();
		this.bizpool.execute(new BizThread(this, obj, pool, thread));
	}

	public synchronized boolean isBlock() throws InterruptedException {
		while (((ThreadPoolExecutor) bizpool).getQueue().size() > maxBlockSize) {
			log.warn("Blocking:: maxBlockSize:" + maxBlockSize + ", active:"
					+ ((ThreadPoolExecutor) bizpool).getActiveCount() + ", block:"
					+ ((ThreadPoolExecutor) bizpool).getQueue().size());
			this.wait();
		}
		return true;
	}

	public synchronized void releaseThread() {
		super.notifyAll();
	}

	public void init() throws Exception {
	}

	public void destroy() {
		this.log.warn("destroy thread pool, isShutdown:" + this.bizpool.isShutdown());
		if (!this.bizpool.isShutdown())
			this.bizpool.shutdown();
	}

	public void setOnMessage(IOnMessage onMessage) {
		this.onMessage = onMessage;
	}

	public void setPoolSize(int size) {
		this.bizpool = Executors.newFixedThreadPool(size);
	}
}
